package com.wang.kafkastudy.component;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSON;
import cn.hutool.json.JSONUtil;
import com.wang.kafkastudy.entity.Message;
import com.wang.kafkastudy.mapper.MessageMapper;
import com.wang.kafkastudy.service.MessageService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @program: KafkaStudy
 * @ClassName KafkaConsumer
 * @description:
 * @author: wang.yang7
 * @date: 2024-07-26 15:39
 * @Version 1.0
 **/
@Component
public class KafkaConsumer {


    @Resource
    private MessageService messageService;

    //监听消费
    @KafkaListener(topics = {"wangyang"})
    public void onNormalMessage(ConsumerRecord<Object, Object> record) {

        if (record.value() instanceof String){
            Message message = new Message();
            message.setMessage(StrUtil.toString(record.value()));
            message.setOffset(StrUtil.toString(record.offset()));
            message.setCreateTime(StrUtil.toString(record.timestamp()));
            message.setStatus(1);
            message.setTopics(record.topic());

            messageService.save(message);
        } else if (record.value() instanceof Message) {
            messageService.save((Message) record.value());
        }

        System.out.println( );
        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "=" +
                record.value());
    }


}

